package flex.cc.job.executor.util;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;

import flex.cc.PublicUtil;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.KettleLogStore;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;

import com.xxl.job.core.log.XxlJobLogger;

public class KettleUtil {

	
	/**
	 * 判断是否有key
	 * @param map 需要判断的集合
	 * @param key 集合的key
	 * @return
	 */
	public static boolean hasKey(Map<String,String>	map, String key) {
		return map.keySet().contains(key);
	}
	
	
	/**
	 * 执行kettle脚本
	 * @param kettleInfo 执行脚本所需要的环境变量
	 * @param path 执行路径
	 * @throws Exception 
	 */
	public static void run(Map<String,String> kettleInfo,String path) throws Exception {

		String startdate = kettleInfo.get("startdate");
		String enddate = kettleInfo.get("enddate");
		if(path.endsWith(".kjb")) {
			runJob(kettleInfo, path);
		}else if(path.endsWith(".ktr")){
			runTrans(kettleInfo, path);
		}else {
			XxlJobLogger.log(">>>>>>>>>>>>>>未找到该脚本<<<<<<<<<<<<<<<< : " + path);
			XxlJobLogger.log(">>>>>>>>>>>>>>时间区间： ("+startdate +" ~ "+ enddate +")<<<<<<<<<<<<<<<< : " );
		}
	}
	
	/**
	 * 执行作业核心代码
	 * @param kjbParams 作业执行所需要的环境变量集合
	 * @param kjbPath 执行作业脚本的路径
	 * @throws KettleException
	 */
	public static void runJob(Map<String,String> kjbParams , String kjbPath) throws KettleException{
			
		String uuid = UUID.randomUUID().toString();
		
		String startDate = kjbParams.get("startdate");
		String endDate = kjbParams.get("enddate");
		XxlJobLogger.log(">>>>>>>>>>>>>>>>同步脚本执行的时间段("+startDate+" ~ "+endDate+")=========");
		XxlJobLogger.log(">>>>>>>>>>>>>>>>KettleJob<<<<<<<<<<<<<<<" + uuid + " 执行开始");
		try {
			KettleEnvironment.init();
			JobMeta jobMeta = new JobMeta(kjbPath, null);
			
			Job job = new Job(null, jobMeta);
			if(kjbParams != null) {
				Set<Entry<String, String>> propList = kjbParams.entrySet();
				for (Iterator<Entry<String, String>> it = propList.iterator();it.hasNext(); ) {
					Entry<String, String> prop = it.next();
					String key = prop.getKey();
					String value = prop.getValue();
					job.setVariable(key,value);
				}
			}
				
			job.start();
			job.waitUntilFinished();
			
			int errors = job.getErrors();
			String[] msgList = KettleLogStore.getAppender().getBuffer(job.getLogChannelId(), false).toString().split("\n\r\n");
			if (errors > 0) {
				XxlJobLogger.log(">>>>>>>>>>>>>>>>KettleJob<<<<<<<<<<<<<<<" + uuid + " 执行失败");
				XxlJobLogger.log(">>>>>>>>>>>>>>>>数据同步异常时间段("+startDate+" ~ "+endDate+")<<<<<<<<<<<<<<<");
				String errMsg=msgList[0];
				throw new KettleException(errMsg);
			} else {
				XxlJobLogger.log(">>>>>>>>>>成功执行时间区间     ("+startDate+" ~ "+endDate+")>>>"+msgList[0]);
				XxlJobLogger.log(">>>>>>>>>>>>>>>>KettleJob<<<<<<<<<<<<<<<" + uuid + " 执行成功");
			}
			
		} catch (KettleException e) {
			XxlJobLogger.log(">>>>>>>>>>>>>>>>执行失败<<<<<<<<<<<<<<<--" + e.toString());
			XxlJobLogger.log(">>>>>>>>>>>>>>>>执行失败异常时间段("+startDate+" ~ "+endDate+")<<<<<<<<<<<<<<<");
			throw new KettleException(e);
		}  
	
	}
	
	
	/**
	 * 执行单独转换的核心代码
	 * @param ktrParams 转换执行所需要的环境变量集合
	 * @param ktrPath 执行转换脚本的路径
	 * @throws KettleException 
	 */
	public static void runTrans(Map<String,String> ktrParams , String ktrPath) throws KettleException {

		String uuid = UUID.randomUUID().toString();
		
		String startDate = ktrParams.get("startdate");
		String endDate = ktrParams.get("enddate");
		XxlJobLogger.log(">>>>>>>>>>>>>>>>同步ktr脚本执行的时间段("+startDate+" ~ "+endDate+")<<<<<<<<<<<<<<<");
		XxlJobLogger.log(">>>>>>>>>>>>>>>>KettleTrans<<<<<<<<<<<<<<< " + uuid + " 开始执行");
		try {
			KettleEnvironment.init();
			TransMeta transMeta = new TransMeta(ktrPath);
			Trans trans = new Trans(transMeta);
			if(ktrParams != null) {
				Set<Entry<String, String>> propList = ktrParams.entrySet();
				for (Iterator<Entry<String, String>> it = propList.iterator();it.hasNext(); ) {
					Entry<String, String> prop = it.next();
					String key = prop.getKey();
					String value = prop.getValue();
					trans.setVariable(key,value);
				}
			}
			
			trans.execute(null);
			trans.waitUntilFinished();
			int errors = trans.getErrors();
			String[] msgList = KettleLogStore.getAppender().getBuffer(trans.getLogChannelId(), false).toString().split("\n\r\n");
			if (errors > 0) {
				XxlJobLogger.log(">>>>>>>>>>>>>>>>KettleTrans<<<<<<<<<<<<<<< " + uuid + " 执行失败");
				XxlJobLogger.log(">>>>>>>>>>>>>>>>数据同步异常时间段("+startDate+" ~ "+endDate+")<<<<<<<<<<<<<<<");
				String errMsg=msgList[0];
				throw new KettleException(errMsg);
			} else {
				XxlJobLogger.log(">>>>>>>>>>成功执行时间区间     ("+startDate+" ~ "+endDate+")>>>"+msgList[0]);
				XxlJobLogger.log(">>>>>>>>>>>>>>>>KettleTrans<<<<<<<<<<<<<<< " + uuid + " 执行成功");
			}
			
			
		} catch (KettleException e) {
			
			XxlJobLogger.log(">>>>>>>>>>>>>>>>执行失败<<<<<<<<<<<<<<<--" + e.toString());
			XxlJobLogger.log(">>>>>>>>>>>>>>>>执行失败异常时间段("+startDate+" ~ "+endDate+")<<<<<<<<<<<<<<<");
		
			throw new KettleException(e);
		} 
		
	}
	
	/**
	 * 
	 * @param time
	 * @return
	 * @throws Exception
	 */
	public static String getTime30_15(String time) throws Exception {
		long offsetTime = 1000 * 60 * 30L;
		long afterOffset = 1000 * 60 * 15;
		String t = getTime(time, offsetTime, afterOffset);
		return t;
	}
	
	
	/**
	 * 根据传入的时间，获取时间区间
	 * @param time 传入的时间
	 * @param offsetTime 传入时间之前偏移量
	 * @param afterOffset  时间跨度
	 * @return 时间区间字符串  
	 * @throws Exception
	 */
	public static String getTime(String time,Long offsetTime,Long afterOffset) throws Exception {
		String currentTime = time;
		int minute ;
		String minuteStr = currentTime.substring(14,16);
		minute = Integer.parseInt(minuteStr);
		String subTime = currentTime.substring(0,14);
		String startTime = "";
		String endTime = "";
		
		String calc;
		long bt = offsetTime;
		long offsetT = afterOffset;
		
		try {
			if ((minute >= 0) && (minute < 15)) {
				calc = subTime + "00:00";
				SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:00");
				long t = sf.parse(calc).getTime() - bt;
				Date st = new Date(t);
				Date et = new Date(t + offsetT);
				SimpleDateFormat sff = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				startTime = sff.format(st);
				endTime = sff.format(et);
			}else if((minute >= 15) && (minute < 30)){
				calc = subTime + "15:00";
				SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:00");
				long t = sf.parse(calc).getTime() - bt;
				Date st = new Date(t);
				Date et = new Date(t + offsetT);
				SimpleDateFormat sff = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				startTime = sff.format(st);
				endTime = sff.format(et);
			}else if((minute >= 30) && (minute < 45)){
				calc = subTime + "30:00";
				SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:00");
				long t = sf.parse(calc).getTime() - bt;
				Date st = new Date(t);
				Date et = new Date(t + offsetT);
				SimpleDateFormat sff = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				startTime = sff.format(st);
				endTime = sff.format(et);
			}else {
				calc = subTime + "45:00";
				SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:00");
				long t = sf.parse(calc).getTime() - bt;
				Date st = new Date(t);
				Date et = new Date(t + offsetT);
				SimpleDateFormat sff = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				startTime = sff.format(st);
				endTime = sff.format(et);
			}
			
			return startTime+","+endTime;
		} catch (ParseException e) {
			XxlJobLogger.log("位置：KettleJob.KettleJob.getTime(String time,Long offsetTime,Long afterOffset)");
			XxlJobLogger.log(e.toString());
			throw new Exception(e);
		}
	}

	
	/**
	 * 获取当前时间的前一天时间区间
	 * @return
	 * @throws Exception 
	 */
	public static String getTime() throws Exception {
		Date d = new Date();
		SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		String currTime = f.format(d);
		String time = getTime(currTime);
		return time;
	}
	
	/**
	 * 获取指定时间的前一天的时间区间
	 * @param time
	 * @return
	 * @throws Exception
	 */
	public static String getTime(String time) throws Exception {
			
			Calendar c = Calendar.getInstance();
			SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
			String currTime = time.substring(0,10);
			Date date = null;
			try {
				date = f.parse(currTime);
			} catch (ParseException e) {
				XxlJobLogger.log(e.toString());
				throw new Exception(e);
			}
			c.setTime(date);
			int day = c.get(Calendar.DATE);
			c.set(Calendar.DATE, day - 1);
			String beforeTime = f.format(c.getTime());
			String startTime = new StringBuffer(beforeTime).append(" 00:00:00").toString();
			String endTime = new StringBuffer(currTime).append(" 00:00:00").toString();
			
			return new StringBuilder(startTime).append(",").append(endTime).toString();
	}
	
	/**
	 * 获取当前时间的前三天时间区间
	 * @return
	 * @throws Exception 
	 */
	public static String getThreeTime() throws Exception {
		Date d = new Date();
		SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		String currTime = f.format(d);
		String time = getThreeTime(currTime);
		return time;
	}
	
	/**
	 * 获取指定时间的前三天的时间区间
	 * @param time
	 * @return
	 * @throws Exception
	 */
	public static String getThreeTime(String time) throws Exception {
		
		Calendar c = Calendar.getInstance();
		SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
		String currTime = time.substring(0,10);
		Date date = null;
		try {
			date = f.parse(currTime);
		} catch (ParseException e) {
			XxlJobLogger.log(e.toString());
			throw new Exception(e);
		}
		c.setTime(date);
		int day = c.get(Calendar.DATE);
		c.set(Calendar.DATE, day - 3);
		String beforeTime = f.format(c.getTime());
		String startTime = new StringBuffer(beforeTime).append(" 00:00:00").toString();
		String endTime = new StringBuffer(currTime).append(" 00:00:00").toString();
		
		return new StringBuilder(startTime).append(",").append(endTime).toString();
	}
	
	
	/**
	 * 获取指定两个时间范围内的时间集合-按每天处理
	 * 例：time1=2020-03-21 00:00:00  time1=2020-03-24 00:00:00
	 * return 2020-03-21 00:00:00,2020-03-22 00:00:00   2020-03-22 00:00:00,2020-03-23 00:00:00  2020-03-23 00:00:00,2020-03-24 00:00:00
	 * @param time1
	 * @param time2
	 * @return
	 * @throws Exception
	 */
	public static List<String> getTimeList(String time1,String time2) throws Exception {
		List<String> list = new ArrayList<>();
		long datePlus = PublicUtil.dateCompare(time1,time2);
		for (int i = 0; i < datePlus; i++) {
			if(PublicUtil.dateCompare(time1,time2)>0) {
				time1 = getAfterTime(time1);
				list.add(time1);
				time1 = time1.split(",")[1];
			}
		}
		return list;
	}
	/**
	 * 获取指定时间的后一天的时间区间
	 * @param time
	 * @return
	 * @throws Exception
	 */
	public static String getAfterTime(String time) throws Exception {
		Calendar c = Calendar.getInstance();
		SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
		String currTime = time.substring(0,10);
		Date date = null;
		try {
			date = f.parse(currTime);
		} catch (ParseException e) {
			XxlJobLogger.log(e.toString());
			throw new Exception(e);
		}
		c.setTime(date);
		int day = c.get(Calendar.DATE);
		c.set(Calendar.DATE, day + 1);
		String afterTime = f.format(c.getTime());
		String startTime = new StringBuffer(currTime).append(" 00:00:00").toString();
		String endTime = new StringBuffer(afterTime).append(" 00:00:00").toString();
		
		return new StringBuilder(startTime).append(",").append(endTime).toString();
	}
}
